package cn.com.core;

import cn.com.constant.ConfigConstant;
import cn.com.util.SqlFileUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.regex.Matcher;


/**
 * 创建flink应用程序
 *
 * @author WZQ
 * @date 2021-08-27 20:58
 */
public class FlinkApplication {

    /**
     * 运行命令的目录
     */
    private static final String COMMEND_RUNNING_PATH;

    private static final Logger LOGGER = LoggerFactory.getLogger(FlinkApplication.class);

    static {
        String path;
        try {
            // 获取命令执行的目录，注意不是jar所在目录
            path = new File("").getCanonicalPath();
        } catch (IOException e) {
            path = "";
            e.printStackTrace();
        }
        if (!path.endsWith(File.separator)) {
            path += File.separator;
        }
        COMMEND_RUNNING_PATH = path;
    }

    /**
     * 初始化表环境、设置基础默认参数、解析sql文件内容，执行sql语句，提交任务
     *
     * @param parameterTool 输入参数
     */
    public void start(ParameterTool parameterTool) {
        List<String> content = SqlFileUtil.getFileContentAsSql(COMMEND_RUNNING_PATH + parameterTool.get(ConfigConstant.SQL_FILE_PATH));
        TableEnvironment tEnv = getTableEnvironment(parameterTool);
        executeAllSql(tEnv, content);
    }

    /**
     * 获取表环境
     *
     * @param parameterTool 输入参数
     * @return 表环境
     */
    private TableEnvironment getTableEnvironment(ParameterTool parameterTool) {
        TableEnvironment tEnv;
        if (Objects.equals(ConfigConstant.STREAMING_MODE, parameterTool.get(ConfigConstant.EXECUTION_MODE).toLowerCase())) {
            EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
            StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            tEnv = StreamTableEnvironment.create(sEnv, settings);
        } else if (Objects.equals(ConfigConstant.BATCH_MODE, parameterTool.get(ConfigConstant.EXECUTION_MODE).toLowerCase())) {
            EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
            tEnv = TableEnvironment.create(settings);
        } else {
            throw new IllegalArgumentException("必须使用streaming或batch模式中的一种，不可使用其他模式。");
        }
        return tEnv;
    }

    /**
     * 按照给定的 sql 语句列表按序执行所有 sql 语句
     *
     * @param tEnv   {@link TableEnvironment}对象
     * @param allSql 所有待执行的 sql 语句
     */
    private void executeAllSql(TableEnvironment tEnv, List<String> allSql) {
        StatementSet statementSet = tEnv.createStatementSet();
        boolean haveInsertSql = false;
        for (String sql : allSql) {
            // set
            Matcher setMatcher = ConfigConstant.SET_PATTERN.matcher(sql);
            if (setMatcher.matches()) {
                setStatement(tEnv.getConfig().getConfiguration(), setMatcher);
                continue;
            }
            // insert
            if (ConfigConstant.INSERT_PATTERN.matcher(sql).matches()) {
                haveInsertSql = true;
                insertStatement(statementSet, sql);
                continue;
            }
            // select
            // if (ConfigConstant.SELECT_PATTERN.matcher(sql).matches()) {
            //     selectStatement(tEnv, sql);
            //     continue;
            // }
            // 其他的所有语句均通过通用的 executeSql 函数来执行
            executeSql(tEnv, sql);
        }
        if (haveInsertSql) {
            // insert 语句需要最后单独调用执行方法去执行
            statementSet.execute();
        }
    }

    /**
     * 将set语句内容设置到底层配置中，一般都是一些优化设置
     *
     * @param conf    配置
     * @param matcher set 语句匹配 matcher
     */
    private void setStatement(Configuration conf, Matcher matcher) {
        conf.setString(matcher.group(1), matcher.group(2));
    }

    /**
     * 将insert语句加入statementSet，包括into和overwrite
     *
     * @param statementSet statementSet
     * @param sql          sql语句：into、overwrite
     */
    private void insertStatement(StatementSet statementSet, String sql) {
        statementSet.addInsertSql(sql);
    }


    /**
     * 执行sql语句，包括：创建表、创建视图等
     *
     * @param tEnv TableEnvironment
     * @param sql  sql语句：DDL/DML/DQL/SHOW/DESCRIBE/EXPLAIN/USE
     */
    private void executeSql(TableEnvironment tEnv, String sql) {
        System.out.println("\n\n\n\n运行sql语句：\n" + sql + "\n的结果为：\n");
        tEnv.executeSql(sql).print();
        System.out.println();
    }

    // 暂时先去掉对 select 结果的单独解析
    // /**
    //  * 执行sql语句，并打印结果数据
    //  *
    //  * @param tEnv TableEnvironment对象
    //  * @param sql  单个select语句
    //  */
    // private void selectStatement(TableEnvironment tEnv, String sql) {
    //     TableResult result = tEnv.executeSql(sql);
    //     ResolvedSchema schema = result.getResolvedSchema();
    //     int columnCount = schema.getColumnCount();
    //     List<String> columnNames = schema.getColumnNames();
    //
    //     StringBuilder resultStr = new StringBuilder();
    //     resultStr.append("\n查询到以下结果数据：\n\n");
    //
    //     separatorAppend(resultStr, columnCount);
    //     columnNames.forEach(
    //             name -> {
    //                 if (name.length() > 30) {
    //                     name = name.substring(0, ConfigConstant.COLUMN_CHARACTER_MAX_COUNT);
    //                 } else {
    //                     name = StringUtils.rightPad(name, ConfigConstant.COLUMN_CHARACTER_MAX_COUNT, "");
    //                 }
    //                 resultStr.append("|").append(name);
    //             });
    //     resultStr.append("|").append("\n");
    //     separatorAppend(resultStr, columnCount);
    //
    //     try (CloseableIterator<Row> it = result.collect()) {
    //         while (it.hasNext()) {
    //             Row row = it.next();
    //             for (int i = 0; i < columnCount; i++) {
    //                 String value = row.getField(i).toString();
    //                 if (value.length() > ConfigConstant.COLUMN_CHARACTER_MAX_COUNT) {
    //                     value = value.substring(0, ConfigConstant.COLUMN_CHARACTER_MAX_COUNT);
    //                 } else {
    //                     value = StringUtils.rightPad(value, ConfigConstant.COLUMN_CHARACTER_MAX_COUNT, "");
    //                 }
    //                 resultStr.append("|").append(value);
    //             }
    //             resultStr.append("|").append("\n");
    //         }
    //         separatorAppend(resultStr, columnCount);
    //         System.out.println("sout输出");
    //         LOGGER.info("logger输出");
    //         LOGGER.info(resultStr.toString());
    //     } catch (Exception e) {
    //         LOGGER.error(ExceptionUtils.stringifyException(e));
    //     }
    // }

    // /**
    //  * 打印分隔符号+----------+
    //  *
    //  * @param resultStr   结果字符串
    //  * @param columnCount 字段数量
    //  */
    // private void separatorAppend(StringBuilder resultStr, int columnCount) {
    //     for (int i = 0; i < columnCount; i++) {
    //         resultStr.append("+------------------------------");
    //     }
    //     resultStr.append("+").append("\n");
    // }

}
